package demo04;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Arrays;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class MovieClassifyMapper extends Mapper<LongWritable, Text, Text, DistanceAndLabel> {
	private DistanceAndLabel distance_label=new DistanceAndLabel();
	private String splitter="";
	ArrayList<String> testData=new ArrayList<String>();
	private String testPath="";
	@Override
	protected void setup(Mapper<LongWritable, Text, Text, DistanceAndLabel>.Context context)
			throws IOException, InterruptedException {
		Configuration conf=context.getConfiguration();
		splitter=conf.get("SPLITTER");
		testPath=conf.get("TESTPATH"); //"/movie/testData"
		//读取测试数据存于ArrayList列表testData中
		FileSystem fs=FileSystem.get(conf);
		FSDataInputStream is=fs.open(new Path(testPath));
		BufferedReader br=new BufferedReader(new InputStreamReader(is));
		String line="";
		while((line=br.readLine())!=null){
			testData.add(line); //100,0,35,17,95401,62,33,0,1,11,6,0,24,2,1,2,1,4,14,24,14,14,3
		}
		is.close();
		br.close();
	}
	//读取训练数据
	@Override
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, DistanceAndLabel>.Context context)
			throws IOException, InterruptedException {
		double distance=0.0;
		String[] val=value.toString().split(splitter);//训练数据 1,1,1,10,48067,5,5,18,20,14,2,0,21,3,0,0,14,0,6,3,3,2,0
		String[] singleTrainData=Arrays.copyOfRange(val, 5, val.length);// 取下标为5 到 val结束的数据 5,5,18,20,14,2,0,21,3,0,0,14,0,6,3,3,2,0
		String label=val[1]; //性别
		for (String td: testData) {
			String[] test=td.split(splitter);
			String[] singleTestData=Arrays.copyOfRange(test, 5, test.length);//测试数据   62,33,0,1,11,6,0,24,2,1,2,1,4,14,24,14,14,3
			distance=Distance(singleTrainData,singleTestData);
			distance_label.setDistance(distance);
			distance_label.setLabel(label);
			context.write(new Text(td), distance_label);	 //key是测试数据	 value是自定义数据DistanceAndLabel
			//100,0,35,17,95401,62,33,0,1,11,6,0,24,2,1,2,1,4,14,24,14,14,3  距离,类别
		}
	}
	/**
	 * 计算训练数据与测试数据的距离  夹角余弦
	 * @param singleTrainData
	 * @param singleTestData
	 * @return
	 */
	private double Distance(String[] singleTrainData, String[] singleTestData) {
		double ab=0.0;
		double aa=0.0;
		double bb=0.0;
//		double TrainSum = 0.0;
//		double TestSum = 0.0;
//		for(int i = 0; i<singleTrainData.length;i++) {
//			TrainSum += Double.parseDouble(singleTrainData[i]); 
//		}
//		for(int i = 0; i<singleTestData.length;i++) {
//			TestSum += Double.parseDouble(singleTrainData[i]); 
//		}
		
		for(int i=0;i<singleTrainData.length;i++){
			ab += (Double.parseDouble(singleTrainData[i]) )*(Double.parseDouble(singleTestData[i]) );
			aa += (Double.parseDouble(singleTrainData[i]) )*(Double.parseDouble(singleTrainData[i]));
			bb += (Double.parseDouble(singleTestData[i])  ) *(Double.parseDouble(singleTestData[i]));
		}
		
		return  1 - ab/(Math.sqrt(aa)*Math.sqrt(bb)) ;
		
	}
}
